package com.meteor.raft.healthcheck;

import com.alibaba.fastjson.JSON;
import com.meteor.raft.constant.CommonConstant;
import com.meteor.raft.core.RaftPeerManage;
import com.meteor.raft.misc.HttpClient;
import com.meteor.raft.misc.UrlUtil;
import com.meteor.raft.persistent.RaftPeer;
import com.meteor.raft.util.NetUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.HashMap;
import java.util.Map;

/**
 * @author gang.wang1903@xiaobao100.cn
 * @Date: 2019/7/19 3:10 PM
 */
@Service
@Slf4j
public class HealthCheckServiceImpl implements HealthCheckService {


    @Autowired
    private RaftPeerManage raftPeerManage;

    @Autowired
    private NetUtils netUtils;

    @Override
    public void sendBeat() {

        RaftPeer local = raftPeerManage.getPeers().get(netUtils.localServer());
        if (local.state != RaftPeer.State.LEADER) {
            return;
        }
        log.info("[RAFT] start  sending beat");
        log.info("[RAFT] send beat with {} keys.", local.getIp());
        local.resetLeaderDue();
        Map<String, String> params = new HashMap<String, String>(1);
        params.put("beat", JSON.toJSONString(local));
        for (final String server : raftPeerManage.allServersIncludeMyself()) {
            final String url = UrlUtil.buildURL(server, CommonConstant.API_BEAT);
            try {
                HttpClient.HttpResult result =  HttpClient.httpPost(url, null, params,"utf-8");
                log.info("received beat from peer: {}", JSON.toJSONString(result.content));

            } catch (Exception e) {
                log.warn("error while sending beat to server: {}", server);
            }
        }


    }

    @Override
    public RaftPeer receivedBeat(RaftPeer beat) {

        final RaftPeer local = raftPeerManage.getPeers().get(netUtils.localServer());
        final RaftPeer remote = new RaftPeer();

        if (beat.state != RaftPeer.State.LEADER) {
            log.info("[RAFT] invalid state from master, state: {}, remote peer: {}", remote.state, JSON.toJSONString(remote));
            throw new IllegalArgumentException("invalid state from master, state: " + remote.state);
        }

        if (local.term.get() > beat.term.get()) {
            log.info("[RAFT] out of date beat, beat-from-term: {}, beat-to-term: {}, remote peer: {}, and leaderDueMs: {}", remote.term.get(), local.term.get(), JSON.toJSONString(remote), local.leaderDueMs);
            throw new IllegalArgumentException("out of date beat, beat-from-term: " + remote.term.get() + ", beat-to-term: " + local.term.get());
        }
        local.resetLeaderDue();
        local.resetHeartbeatDue();
        raftPeerManage.makeLeader(beat);
        return local;
    }
}
